package com.shujia.traffic.dws

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DwsAppVioWfjlgdDay {
  def main(args: Array[String]): Unit = {
    val bizdate: String = "20250111"

    // 创建SparkSession
    // 如果不在本地执行，则master及config配置在提交任务时统一指定，代码中不写死
    val spark: SparkSession = SparkSession
      .builder()
      .appName("DwsAppVioWfjlgdDay")
      .enableHiveSupport() // 开启Hive的支持，使Spark代码能够获取Hive的表及数据
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val vioDF: DataFrame = spark.table("dwd.dwd_base_vio_surveil_msk_d_f").where($"pt" === bizdate)

    // 统计每个设备每一天拍摄的违法数量
    val resDF: DataFrame = vioDF
      .groupBy($"sbbh", date_format($"wfsj", "yyyy-MM-dd") as "wfsj_day")
      .agg(count("*") as "sbwfs_day")
      // 过滤掉sbbh异常以及时间日期异常的数据
      .where($"sbbh".isNotNull and trim($"sbbh") =!= "" and $"wfsj_day".isNotNull)
      // 基于箱线图原理，根据每个设备每一天的违法记录数进行统计
      .withColumn("rn", row_number() over Window.partitionBy($"sbbh").orderBy($"sbwfs_day"))
      .withColumn("cnt", count("*") over Window.partitionBy($"sbbh"))
      // 计算Q1、Q3的位置
      .withColumn("Q1_pos", floor($"cnt" / 4))
      .withColumn("Q3_pos", ceil($"cnt" * 3 / 4))
      // 找到Q1、Q3的值
      .withColumn("Q1", sum(when($"rn" === $"Q1_pos", $"sbwfs_day").otherwise("0")) over Window.partitionBy($"sbbh"))
      .withColumn("Q3", sum(when($"rn" === $"Q3_pos", $"sbwfs_day").otherwise("0")) over Window.partitionBy($"sbbh"))
      .withColumn("IQR", $"Q3" - $"Q1")
      // 计算上边界
      .withColumn("zdgjz", $"Q3" + $"IQR" * 1.5)
      .where($"sbwfs_day" > $"zdgjz")
      .select($"sbbh", $"sbwfs_day", $"zdgjz", $"wfsj_day")
      // 保留每个设备日期最大的一条记录
      .withColumn("rn", row_number() over Window.partitionBy($"sbbh").orderBy($"wfsj_day".desc))
      .where($"rn" === 1)
      .select($"sbbh", $"sbwfs_day", $"zdgjz", $"wfsj_day")

    spark.sql(
      s"""
        |alter table dws.dws_app_vio_wfjlgd_d add if not exists partition(pt='$bizdate')
        |""".stripMargin)

    // 将结果保存到dws的结果表中
    resDF
      .write
      .mode(SaveMode.Overwrite)
      .format("orc")
      .save(s"/project/data/traffic/dws/dws_app_vio_wfjlgd_d/pt=$bizdate")

  }

}
